package com.simple.mq.boot.starter;

import com.simple.mq.boot.starter.annotation.MqConsumer;
import com.simple.mq.boot.starter.config.SimpleMqCommonConfig;
import com.simple.mq.boot.starter.service.SimpleMqConsumerService;
import com.simple.mq.common.config.SimpleThreadPool;
import com.simple.mq.sdk.consumer.SimpleMqConsumer;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

import javax.annotation.Resource;
import java.net.URI;
import java.util.Arrays;
import java.util.Objects;

/**
 * 项目: simple-mq
 *
 * 功能描述:
 *
 * @author: WuChengXing
 *
 * @create: 2023-09-24 09:55
 **/
public class ConsumerInitEvent implements ApplicationListener<ContextRefreshedEvent> {

    @Resource
    private SimpleMqCommonConfig simpleMqCommonConfig;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        ApplicationContext applicationContext = event.getApplicationContext();
        String[] beanNamesForAnnotation = applicationContext.getBeanNamesForAnnotation(MqConsumer.class);
        if (beanNamesForAnnotation.length < 1) {
            return;
        }
        SimpleMqConsumerService bean = applicationContext.getBean(SimpleMqConsumerService.class);
        // 判断是否是被@SimpleRpcService注解的类
        MqConsumer mqConsumer = bean.getClass().getAnnotation(MqConsumer.class);
        if (Objects.isNull(mqConsumer)) {
            // 不存在该注解不操作
            return;
        }
        String[] strings = mqConsumer.queueNames();
        String clientName = mqConsumer.clientName();
        SimpleMqConsumer simpleMqConsumer = new SimpleMqConsumer(URI.create(transfer()), clientName);
        simpleMqConsumer.register(Arrays.asList(strings), true);
        SimpleThreadPool.INSTANCE.submit(new ProcessConsumer(simpleMqConsumer, bean));
    }

    private String transfer() {
        return "ws://" + simpleMqCommonConfig.getHost() + ":" + simpleMqCommonConfig.getPort() + "/";
    }

    static class ProcessConsumer implements Runnable {

        private final String queueName;
        private final SimpleMqConsumer simpleMqConsumer;
        private final Object bean;

        public ProcessConsumer(SimpleMqConsumer simpleMqConsumer, Object bean) {
            this.simpleMqConsumer = simpleMqConsumer;
            this.bean = bean;
            this.queueName = Thread.currentThread().getName();
        }

        public ProcessConsumer(SimpleMqConsumer simpleMqConsumer, Object bean, String queueName) {
            this.simpleMqConsumer = simpleMqConsumer;
            this.bean = bean;
            this.queueName = queueName;
        }

        @Override
        public void run() {
            SimpleMqConsumerService service = (SimpleMqConsumerService) bean;
            simpleMqConsumer.onMessage((msg) -> service.consumer(queueName, msg));
        }
    }
}
